package com.jiayuan.cn.energy.test.one;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.core.execution.JobClient;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
//import org.apache.flink.table.descriptors.Json;
import org.apache.flink.table.factories.TableSourceFactory;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Optional;

public class AppSqlMqtt {
    public static void main(String[] args) throws Exception {

//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        Configuration configuration = new Configuration();
        configuration.setString(RestOptions.BIND_PORT, "8081");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration);
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        tEnv.executeSql("""
                CREATE TABLE mqtt_word (
                    data STRING
                ) WITH (
                    'connector' = 'word'
                )
                """);
//        ,
//        'connector.topic' = '/v1/report/+/model',
//                'connector.broker_url' = 'tcp://192.168.110.88:1883',
//                'connector.username' = 's3',
//                'connector.password' = '111',
//                'connector.client_id' = 'S916',
//                'format' = 'delimited',
//                'format.field-delimiter' = ','
//        'connector.type' = 'com.jiayuan.cn.energy.test.one.MQSource',
        // Use the MQTT source in Flink SQL queries

        String dataPrint= """
                create table data_print(
                str STRING
                )
                WITH (
                'connector' = 'print'
                )
                """;
        tEnv.executeSql(dataPrint);
        TableResult tableResult =   tEnv.executeSql("insert into data_print SELECT data as str FROM mqtt_word");
        Optional<JobClient> jobClient = tableResult.getJobClient();
        jobClient.ifPresent(client->{System.out.println("Energy job id ::"+client.getJobID());});

    }
}